package com.cn.kafka;

import java.io.IOException;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

import org.apache.kafka.common.KafkaException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import com.cn.service.OrderService;
@Component
public class KafkaSender {
	 @Resource
	    KafkaConsumerPool consumerPool;
		@Autowired
		private OrderService orderService;

	    /**
	     *  这里需要放到程序启动完成之后执行 TODO
	     */
	    @PostConstruct
	    void d(){

	        ConsumerGroup consumerThread = new ConsumerGroup("order-1","order_list",consumerConfig);
	        ConsumerGroup consumerThread2 = new ConsumerGroup("order-2","order_list", consumerConfig);

	        /**
	         * 各起两个消费者 ,Kafka consumer是非线程安全的 Consumer 需要一个new 的
	         */
	        consumerPool.SubmitConsumerPool(new Consumer(consumerThread,orderService));
	        consumerPool.SubmitConsumerPool(new Consumer(consumerThread,orderService));

	        //consumerPool.SubmitConsumerPool(new Consumer(consumerThread2,orderService));
	       // consumerPool.SubmitConsumerPool(new Consumer(consumerThread2,orderService));
	    }


	    @Resource
	    KafkaConsumerConfig consumerConfig;

	    @Autowired
	    private KafkaTemplate kafkaTemplate;

 

	    /**
	     * 发送消息到kafka
	     *
	     */
	    public void sendTest() throws InterruptedException, IOException, KafkaException {

	        /**
	         * topic='access_data'
	         */
	    	System.out.println("send message");
	        kafkaTemplate.send("access_data",""+ System.currentTimeMillis());
	        kafkaTemplate.send("access_data",""+System.currentTimeMillis());
	        kafkaTemplate.send("access_data",""+System.currentTimeMillis());
	        kafkaTemplate.send("access_data",""+System.currentTimeMillis());
	        kafkaTemplate.send("access_data",""+System.currentTimeMillis());
	        kafkaTemplate.send("access_data",""+System.currentTimeMillis());
	      
	    }
	    public void sendTopicsMessage(String topic,String message) throws InterruptedException, IOException, KafkaException {

	        /**
	         * topic='access_data'
	         */
	    	System.out.println("send message");
	        kafkaTemplate.send(topic,message);
	        System.out.println("send message over");
	    }
}
